from pyspark import SparkConf, SparkContext


if __name__ == "__main__":
    conf = SparkConf().setMaster("local").setAppName("quzhi")
    sc = SparkContext(conf=conf)
    lines = sc.textFile("58data.txt")

    def GetLine(line):
        infos =line.split("\t")
        return infos[0],infos[1],infos[2],infos[3],infos[4],int(infos[5]),infos[6]

    pairRDD = lines.map(lambda line:GetLine(line))

    #按照价格排序
    sort = pairRDD.sortBy(lambda x: x[5], True)
    f = open("zfSortByPrice.txt", "w")
    for line in sort.collect():
        f.write(','.join(map(str, line)))
        f.write('\n')
    f.close()

'''
    #读取价格最便宜的前10条数据
    GetData = sort.take(10)
    [print(x) for x in GetData]

    #筛选价格大于1000的数据
    filterRDD = pairRDD.filter(lambda x: x[5] > 1000)
    reasultRDD = filterRDD.collect()
    [print(x) for x in reasultRDD]

    #单间和整租平均价格
    def GetLine1(line):
        infos =line.split("\t")
        return infos[1],int(infos[5])
    pairRDD1 = lines.map(lambda line: GetLine1(line))
    createCombiner = (lambda value: (value, 1))
    mergeValue = (lambda merge1, sale: (merge1[0] + sale, merge1[1] + 1))
    mergeGroup = (lambda merge1, merge2: (merge1[0] + merge2[0], merge1[1] + merge2[1]))
    data = pairRDD1.combineByKey(createCombiner, mergeValue, mergeGroup)
    print(data.collect())
    def average(x):
        Type = x[0]
        price = x[1][0]
        count = x[1][1]
        return (Type, price / count)
    result = data.map(lambda x: average(x))
    result.foreach(print)


    #单间和整租平均价格
    def GetLine2(line):
        infos =line.split("\t")
        return infos[2],int(infos[5])
    pairRDD2 = lines.map(lambda line: GetLine2(line))
    createCombiner = (lambda value: (value, 1))
    mergeValue = (lambda merge1, sale: (merge1[0] + sale, merge1[1] + 1))
    mergeGroup = (lambda merge1, merge2: (merge1[0] + merge2[0], merge1[1] + merge2[1]))
    data = pairRDD2.combineByKey(createCombiner, mergeValue, mergeGroup)
    print(data.collect())
    def average(x):
        Type = x[0]
        price = x[1][0]
        count = x[1][1]
        return (Type, price / count)
    result = data.map(lambda x: average(x))
    result.foreach(print)'''